package com.fanxuankai.boot.canal.listener;

import com.fanxuankai.boot.canal.listener.annotation.CanalListener;
import com.fanxuankai.boot.canal.listener.annotation.Delete;
import com.fanxuankai.boot.canal.listener.annotation.Insert;
import com.fanxuankai.boot.canal.listener.annotation.Update;
import com.fanxuankai.boot.canal.util.Topic;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author fanxuankai
 */
class CanalListenerDefinition {
    private final CanalListener canalListener;
    private final String topic;
    private final TableInfo tableInfo;
    private Consumer<Object> insertConsumer;
    private BiConsumer<Object, Object> updateConsumer;
    private Consumer<Object> deleteConsumer;

    public CanalListenerDefinition(Object object, CanalTableCache canalTableCache) {
        // 如果该类在事务管理，会被代理，将拿不到注解，需要借助于官方的工具类
        CanalListener canalListener = AnnotationUtils.findAnnotation(object.getClass(), CanalListener.class);
        assert canalListener != null;
        this.canalListener = canalListener;
        this.tableInfo = canalTableCache.getTableInfo(canalListener.entityClass());
        this.topic = Optional.of(canalListener.topic())
                .filter(StringUtils::hasText)
                .map(Topic::custom)
                .orElse(Topic.of(tableInfo.getSchema(), tableInfo.getTable()));
        for (Method method : object.getClass().getDeclaredMethods()) {
            method.setAccessible(true);
            if (AnnotationUtils.findAnnotation(method, Insert.class) != null) {
                this.insertConsumer = o -> {
                    try {
                        method.invoke(object, o);
                    } catch (IllegalAccessException | InvocationTargetException e) {
                        throw new RuntimeException(e);
                    }
                };
            } else if (AnnotationUtils.findAnnotation(method, Update.class) != null) {
                this.updateConsumer = (o, o2) -> {
                    try {
                        method.invoke(object, o, o2);
                    } catch (IllegalAccessException | InvocationTargetException e) {
                        throw new RuntimeException(e);
                    }
                };
            } else if (AnnotationUtils.findAnnotation(method, Delete.class) != null) {
                this.deleteConsumer = o -> {
                    try {
                        method.invoke(object, o);
                    } catch (IllegalAccessException | InvocationTargetException e) {
                        throw new RuntimeException(e);
                    }
                };
            }
        }
    }

    public String getTopic() {
        return topic;
    }

    public TableInfo getTableInfo() {
        return tableInfo;
    }

    @SuppressWarnings("unchecked")
    public Class<Object> getEntityClass() {
        return (Class<Object>) canalListener.entityClass();
    }


    public Consumer<Object> getInsertConsumer() {
        return insertConsumer;
    }


    public Consumer<Object> getDeleteConsumer() {
        return deleteConsumer;
    }


    public BiConsumer<Object, Object> getUpdateConsumer() {
        return updateConsumer;
    }
}
